/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 *
 */
package org.apache.asyncweb.server.pipeline;

import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.asyncweb.common.HttpResponse;
import org.apache.asyncweb.server.HttpServiceContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class StandardRequestPipeline implements RequestPipeline {

	private static final Logger LOG = LoggerFactory
			.getLogger(StandardRequestPipeline.class);

	private Runnable emptyCommand;

	private int maxPipelinedRequests;

	private RequestPipelineListener listener;

	private Map<HttpServiceContext, HttpResponse> entryMap = new LinkedHashMap<HttpServiceContext, HttpResponse>();

	private ReentrantLock lock = new ReentrantLock();

	public StandardRequestPipeline(int maxPipelinedRequests) {
		super();
		this.maxPipelinedRequests = maxPipelinedRequests;
	}

	public boolean addRequest(HttpServiceContext context) {
		boolean added = false;
		lock.lock();
		try {
			if (entryMap.size() < maxPipelinedRequests) {
				entryMap.put(context, null);
				added = true;
			}
		} finally {
			lock.unlock();
		}

		if (added && LOG.isDebugEnabled()) {
			LOG.debug("Request added to pipeline ok");
		}
		return added;
	}

	public void releaseResponse(HttpServiceContext context) {
		if (context.getCommittedResponse() == null) {
			throw new IllegalStateException("response is not committed.");
		}
		lock.lock();
		try {
			entryMap.put(context, context.getCommittedResponse());
			releaseRequests();
		} finally {
			lock.unlock();
		}
	}

	public void disposeAll() {
		lock.lock();
		try {
			entryMap.clear();
		} finally {
			lock.unlock();
		}
	}

	public void runWhenEmpty(Runnable command) {
		lock.lock();
		try {
			emptyCommand = command;
			if (entryMap.isEmpty()) {
				command.run();
			}
		} finally {
			lock.unlock();
		}
	}

	/**
	 * set the pipeline listener associated with this pipeline
	 * 
	 * @param listener
	 *            The listener
	 */
	public void setPipelineListener(RequestPipelineListener listener) {
		this.listener = listener;
	}

	/**
	 * Releases any requests which can be freed as a result of a request being
	 * freed. We simply iterate through the list (in insertion order) - freeing
	 * all responses until we arive at one which has not yet been completed
	 */
	private void releaseRequests() {
		Map.Entry<HttpServiceContext, HttpResponse> entry;
		for (Iterator<Map.Entry<HttpServiceContext, HttpResponse>> iter = entryMap
				.entrySet().iterator(); iter.hasNext();) {
			entry = iter.next();
			if (entry.getValue() != null) {
				if (LOG.isDebugEnabled()) {
					LOG.debug("Response freed from pipeline. Notifying");
				}
				listener.responseReleased(entry.getKey());
				iter.remove();
			} else {
				break;
			}
		}
		if (emptyCommand != null && entryMap.isEmpty()) {
			emptyCommand.run();
		}
	}

}
